package com.anmi.transfer.schedule;

import com.anmi.transfer.service.TransferCaseOperationService;
import com.anmi.transfer.task.MysqlTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class CaseOperationSchedule {

  @Autowired private TransferCaseOperationService transferCaseOperationService;

  @Value("${fetch.size}")
  private int FETCH_SIZE;

  /** case_operation 表数据同步至ES，这里暂时使用IndexService的异步方法，可以考虑使用Bulk API优化 */
  //  @Scheduled(cron = "0 30 23 * * ?")
  // 全量同步
  public String syncMysqlToES() throws InterruptedException {
    String resultStr = "";
    log.info("定时任务开始启动");
    ExecutorService executorService = null;
    try {
      long start = System.currentTimeMillis();
      Integer dataCount = transferCaseOperationService.countByMap();
      int round = dataCount % FETCH_SIZE != 0 ? dataCount / FETCH_SIZE + 1 : dataCount / FETCH_SIZE;
      // new的线程数量
      MysqlTaskService[] taskServices = new MysqlTaskService[round];
      for (int i = 0; i < round; i++) {
        taskServices[i] = new MysqlTaskService(i * FETCH_SIZE, FETCH_SIZE);
      }
      executorService = Executors.newFixedThreadPool(2);
      Integer[] arr =
          executorService
              .invokeAll(Arrays.asList(taskServices))
              .parallelStream()
              .map(
                  temp -> {
                    try {
                      return temp.get(1, TimeUnit.MINUTES);
                    } catch (Exception e) {
                      log.error("获取任务结果失败,{}", e.getMessage());
                      return 0;
                    }
                  })
              .toArray(Integer[]::new);
      int total = 0;
      for (Integer i : arr) {
        total += i;
      }
      long timeCoast = System.currentTimeMillis() - start;
      log.info("总共同步数据{}条,共使用时间{}ms", total, timeCoast);
      resultStr = String.format("总共同步数据%d条,共使用时间%dms", total, timeCoast);
    } catch (RuntimeException e) {
      log.error(e.getMessage());
    } finally {
      if (executorService != null) executorService.shutdown();
    }
    return resultStr;
  }
}
